package com.gitee.loyo;

import com.gitee.loyo.utils.HttpUtils;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.http.*;
import io.vertx.core.http.impl.headers.HeadersMultiMap;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import static com.gitee.loyo.Const.REQUEST_RESULT_HEADER;
import static com.gitee.loyo.Const.WEBSOCKET_CLIENT_HEADER;

public class BridgeClient extends AbstractVerticle {
    final Logger log = LoggerFactory.getLogger(BridgeClient.class);
    final WebSocketConnectOptions webSocketConnectOptions;
    final JsonObject proxyOptions;
    HttpClientOptions proxyHttpOptions;
    HttpClientOptions wsHttpOptions;
    HttpClient proxyClient;
    HttpClient wsClient;
    Map<String, RequestReady> unanswered = new ConcurrentHashMap<>();
    long lastHeartbeatTime = System.currentTimeMillis();
    Long reconnectTimer;

    public BridgeClient(JsonObject proxyOptions) {
        JsonObject bridgeConfig = proxyOptions.getJsonObject("bridge", new JsonObject());
        this.webSocketConnectOptions = new WebSocketConnectOptions(bridgeConfig.getJsonObject("client", new JsonObject()));
        this.proxyOptions = bridgeConfig.getJsonObject("proxy", new JsonObject());
    }

    @Override
    public void start(Promise<Void> startPromise) throws Exception {
        String hostname = webSocketConnectOptions.getHeaders() == null ? null : webSocketConnectOptions.getHeaders().get(WEBSOCKET_CLIENT_HEADER);
        if(hostname == null || hostname.isEmpty()){
            startPromise.fail("required property: bridge.client.headers." + WEBSOCKET_CLIENT_HEADER);
            return;
        }
        // encode client id
        String header = hostname;
        webSocketConnectOptions.putHeader(WEBSOCKET_CLIENT_HEADER, Base64.getEncoder().encodeToString(header.getBytes(StandardCharsets.UTF_8)));

        proxyHttpOptions = new HttpClientOptions(proxyOptions);
        proxyClient = vertx.createHttpClient(proxyHttpOptions).connectionHandler(conn -> {
            log.info("new proxy conn {}", conn);
            conn.exceptionHandler(t -> log.error("proxy conn error", t))
                .closeHandler(event -> log.info("close proxy conn {}", conn));
        });

        final HttpClientOptions defaultHttpOptions = new HttpClientOptions();
        defaultHttpOptions.setConnectTimeout(5000);
        wsHttpOptions = new HttpClientOptions(defaultHttpOptions.toJson().mergeIn(config(), true));

        connectServer();
        startPromise.complete();
    }

    private synchronized void connectServer(){
        if(wsClient != null){
            wsClient.close(closeEvent -> {
                if(closeEvent.failed()){
                    log.error("can not close bridge client, please reboot application", closeEvent.cause());
                }else{
                    connect();
                }
            });
        }else{
            connect();
        }

    }

    private void connect(){
        log.info("connecting {}", wsHttpOptions.getDefaultHost(), wsHttpOptions.getDefaultPort());
        wsClient = vertx.createHttpClient(wsHttpOptions);
        wsClient.webSocket(webSocketConnectOptions, wsResult -> {
            if(wsResult.failed()){
                log.error("failed to connect to {}:{}", wsHttpOptions.getDefaultHost(), wsHttpOptions.getDefaultPort());
                reconnect();
                return;
            }

            log.info("API-Bridge client connect to {}:{}", wsHttpOptions.getDefaultHost(), wsHttpOptions.getDefaultPort());
            WebSocket ws = wsResult.result();
            long heartbeatTimer = vertx.setPeriodic(10 * 1000, event -> {
                ws.writeFrame(WebSocketFrame.pingFrame(Buffer.buffer((byte)0))).onComplete(ar -> {
                    if(ar.failed()){
                        log.error("heartbeat error", ar.cause());
                        reconnect();
                    }else if(System.currentTimeMillis() - lastHeartbeatTime > 30 * 1000){
                        log.warn("bridge client connection timeout");
                        reconnect();
                    }
                });
            });
            ws.frameHandler(frame ->
                handleFrame(ws, frame)
            ).closeHandler(avoid -> {
                vertx.cancelTimer(heartbeatTimer);
                unanswered.clear();
                reconnect();
            });
        });
    }

    private synchronized void reconnect() {
        log.info("reconnect after one second");
        try{
            cancelReconnectTimer();
            reconnectTimer = vertx.setTimer(1000, $ -> connectServer());
        }catch (Exception e){
            log.error("failed to set timer for connect server", e);
            cancelReconnectTimer();
        }
    }

    private void cancelReconnectTimer(){
        if(reconnectTimer != null){
            vertx.cancelTimer(reconnectTimer);
            reconnectTimer = null;
        }
    }

    private void handleFrame(WebSocket ws, WebSocketFrame frame) {
        if(frame.type() == WebSocketFrameType.PONG){
            lastHeartbeatTime = System.currentTimeMillis();
            return;
        }
        if(frame.type() != WebSocketFrameType.BINARY){
            return;
        }
        Buffer pack = frame.binaryData();
        String requestId = HttpUtils.unwrapRequestId(pack);
        RequestReady requestReady = unanswered.computeIfAbsent(requestId, s -> new RequestReady(proxyClient));
        HttpUtils.unwrapRequestPack(
            frame.binaryData(),
            httpMethod -> requestReady.setHttpMethod(httpMethod),
            uri -> requestReady.setUri(uri),
            header -> requestReady.addHeader(header.getKey(), header.getValue()),
            data -> requestReady.sendChunk(vertx.getOrCreateContext(), Buffer.buffer(data)),
            end -> requestReady.end(respResult -> {
                unanswered.remove(requestId);
                if(respResult.failed()){
                    respResult.cause().printStackTrace();
                    Future.<Void>future(p ->
                        ws.write(HttpUtils.wrapCode(requestId, 500, "client error: " + respResult.cause().getMessage())).onComplete(p)
                    ).compose(unused ->
                        ws.write(HttpUtils.wrapHeader(requestId, new HeadersMultiMap().add(REQUEST_RESULT_HEADER, "failed")))
                    ).onComplete(ar -> {
                        if(ar.succeeded()){
                            ws.write(HttpUtils.wrapEnd(requestId));
                        }else{
                            ar.cause().printStackTrace();
                        }
                    });
                }else{
                    HttpClientResponse response = respResult.result();
                    response.pause();
                    int code = response.statusCode();
                    String message = response.statusMessage();
                    Future.<Void>future(p ->
                        ws.write(HttpUtils.wrapCode(requestId, code, message)).onComplete(p)
                    ).compose(unused ->
                        ws.write(HttpUtils.wrapHeader(requestId, response.headers().add(REQUEST_RESULT_HEADER, "succeed")))
                    ).onComplete(ar -> {
                        if(ar.succeeded()){
                            response.exceptionHandler(t ->
                                t.printStackTrace()
                            ).endHandler(avoid ->
                                ws.write(HttpUtils.wrapEnd(requestId))
                            ).handler(data ->
                                ws.write(HttpUtils.wrapChunk(requestId, data))
                            );
                        }else{
                            ar.cause().printStackTrace();
                        }
                        response.resume();
                    });
                }
            }),
            close -> {
                log.warn("Request " + requestId + " closed");
                requestReady.close();
            }
        );
    }
}
